//package org.example
//
//import org.apache.spark.SparkContext
//
///**
// * 将Spark作业统计的数据保存到redis中
// * @author xingzhiyan
// * @date 2021/4/22 16:43
// */
//class RedisProcessor {
//
//  def save2redis(sc:SparkContext, partitionNum:Int, resultPath:String, redisKey:String, maxSaveValueLen:Int): Unit ={
//    val resrdd = sc.textFile(resultPath)
//      .map(_.split("\t"))
//      .filter(_.length == 2)
//      .map(v=> (v(0), v(1).split(",")))
//      .filter(_._2.nonEmpty)
//      .map(v=> (v._1, v._2.toList))
//
//    resrdd.repartition(partitionNum).foreachPartition(p => {
//      val jedisCluster = JedisFactory.getClient
//
//      val jcp: JedisClusterPipeline = JedisClusterPipeline.pipelined(jedisCluster)
//      try {
//        // batch write
//        while (p.hasNext) {
//          val rsPart = p.next()
//          val rr: Array[String] = rsPart._2.toArray[String]
//          jcp.lpush(redisKey + rsPart._1, rr: _*)
//          jcp.ltrim(redisKey + rsPart._1,0, maxSaveValueLen)
//          jcp.expire(redisKey + rsPart._1, 1)
//        }
//        jcp.sync()
//        true
//      } catch {
//        case e: Exception => {
//        }
//          false
//      } finally {
//        jcp.close()
//      }
//    })
//  }
//
//}
